package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.text.SimpleDateFormat;
import java.util.Properties;
import java.util.TimeZone;

public class test4 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //读取kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test3");
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));
        //输出
        //stream.print();

        //设置水位线
        SingleOutputStreamOperator<String> filterDS = stream.filter(x -> {
            return (JSON.parseObject(x).getString("table").equals("transport_task") &&
                    JSON.parseObject(x).getJSONObject("data").getString("status").equals("67004"));
        });
        //filterDS.print();

        //（4）实时数仓DWD层-业务从原始业务数据库的transport_task表中提取运单完成操作，写入Kafka指定主题。环境准备，
        // 从 Kafka tms_ods 主题读取数据，筛选运输完成数据，处理运输完成数据，获取修改后的数据并转换数据类型，
        // 补全时间戳字段，补全运输时长字段，处理时区问题，脱敏，写出到 Kafka tms_dwd_trans_trans_finish 主题物流域运输完成事实主题（5分）
        SingleOutputStreamOperator<String> mapDS2 = filterDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                TransportTask tt = JSON.parseObject(JSON.parseObject(value).getJSONObject("data").toJSONString(), TransportTask.class);
                //处理时区问题
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                long time = sdf.parse(tt.getCreate_time()).getTime();
                //补全时间戳字段
                tt.setCreate_ts(time);
                //补全运输时长字段
                tt.setTransport_time(sdf.parse(tt.getActual_end_time()).getTime() - sdf.parse(tt.getActual_start_time()).getTime());
                //脱敏
                tt.setDriver1_name(tt.getDriver1_name() != null ? tt.getDriver1_name().substring(0, 1) + "**" : tt.getDriver1_name());
                tt.setDriver2_name(tt.getDriver2_name() != null ? tt.getDriver2_name().substring(0, 1) + "**" : tt.getDriver2_name());

                tt.setTruck_no(tt.getTruck_no() != null ? tt.getTruck_no().substring(0, 3) + "****" : tt.getTruck_no());

                return JSONObject.toJSONString(tt);
            }
        });
        mapDS2.print();
        //{"actual_distance":"226.00","actual_end_time":"1657574157000","actual_start_time":"1657562410000",
        // "create_time":"2022-07-11T21:50:01Z","create_ts":1657547401000,"driver1_emp_id":"1656","driver1_name":"乐**",
        // "end_org_id":"31","end_org_name":"青海省西宁市转运中心","id":"96214","is_deleted":"0","line_id":"844","order_num":"0",
        // "shift_id":"1830","start_org_id":"30","start_org_name":"甘肃省兰州市转运中心","status":"67004",
        // "transport_time":11747000,"truck_id":"915","truck_no":"青DTUETH","update_time":"2022-07-12T02:15:57Z"}
        mapDS2.addSink(new FlinkKafkaProducer<String>("tms_dwd_trans_trans_finish",
                new SimpleStringSchema(), properties));



        env.execute();
    }
}
